package net.opentsdb.aggregation;

import com.google.common.collect.Lists;
import com.google.common.collect.PeekingIterator;
import net.opentsdb.core.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 相邻数据聚合函数处理
 *
 */
public class AggregationAdjacentIterator implements SeekableView, DataPoint, Aggregator.Longs, Aggregator.Doubles {

    private static final Logger LOG = LoggerFactory.getLogger(AggregationIterator.class);

    private final Aggregator aggregator;

    private final boolean rate;

    protected final long start_time;

    protected final long end_time;

    protected long timestamp;

    protected final List<Double> values;

    private DataPoint current;

    private PeekingIterator<DataPoint> source;

    private Iterator<Double> iteratorValues;

    public AggregationAdjacentIterator(final PeekingIterator<DataPoint> source, final long start_time, final long end_time, final Aggregator aggregator,
                               final Aggregators.Interpolation method, final boolean rate) {
        LOG.debug("Aggregating adjacent {} iterators", aggregator.toString());
        this.source = source;
        this.start_time = start_time;
        this.end_time = end_time;
        this.aggregator = aggregator;
        this.rate = rate;
        timestamp = 0L;
        values = Lists.newArrayListWithCapacity(2);

        iteratorValues = values.iterator();
    }


    public boolean hasNext() {
        return source.hasNext();
    }

    public DataPoint next() {
        if(hasNext()){
            current = source.next();
            //清理list数据
            values.clear();
            values.add(current.isInteger()?current.toDouble():current.doubleValue());
            // 预读迭代器数据存入
            if(source.hasNext()){
                DataPoint peek = source.peek();
                values.add(peek.isInteger()?peek.toDouble():peek.doubleValue());
                timestamp = peek.timestamp();
            }else {
                timestamp = current.timestamp();
            }

            iteratorValues = values.iterator();
        }

        return this;
    }

    public void remove() {
        throw new UnsupportedOperationException();
    }

    // 不知作何用
    public void seek(final long timestamp) { }

    public long timestamp() {
        return timestamp;
    }

    public boolean isInteger() {
        return false;
    }

    public long longValue() {
        throw new ClassCastException("current value is a double: " + this);
    }

    public double doubleValue() {
        if(hasNext()){
            final double value = aggregator.runDouble(this);
    //        final double value = current.isInteger()?current.toDouble():current.doubleValue();
            if (Double.isInfinite(value)) {
                throw new IllegalStateException("Got Infinity: " + value + " in this " + this);
            }
            return value;
        }
        return Double.NaN;
    }

    public double toDouble() {
        return isInteger() ? longValue() : doubleValue();
    }

    // -------------------------- //
    // Aggregator.Longs interface //
    // -------------------------- //

    public boolean hasNextValue() {
        return iteratorValues.hasNext();
    }

    public long nextLongValue() {
       return Long.MAX_VALUE;
    }

    // ---------------------------- //
    // Aggregator.Doubles interface //
    // ---------------------------- //

    public double nextDoubleValue() {
        if(hasNextValue()){
            return iteratorValues.next();
        }

        return Double.NaN;
    }

    public String toString() {
        return "SpanGroup.Iterator(timestamps=" + timestamp +
                ", values=" + Arrays.toString(values.toArray()) +
                ", current=" + current +
                ", (SpanGroup: " + toStringSharedAttributes() +
                "), iterators=" + source + ')';
    }

    private String toStringSharedAttributes() {
        return "start_time=" + start_time +
                ", end_time=" + end_time +
                ", rate=" + rate +
                ", aggregator=" + aggregator + ')';
    }

    @Override
    public long valueCount() {
        return values.size();
    }
}
